use std::any::Any;

use hashbrown::HashMap;
use xactor::{Actor, Addr, Context, Handler, Message, Result, Sender, Service};

use crate::actors::{SendMsg, TcpStreamActor};
use crate::message::{HoneybeeMessage, RpcResponse, RpcResponseError, RpcResult};
use crate::xactor_ext::{RpcAddrExt, RpcCaller};

#[derive(Default)]
pub struct Router {
    subscribes: HashMap<u16, Box<dyn Any + Send>>,
}

impl Service for Router {}

impl Actor for Router {
    fn name(&self) -> &str {
        "Router"
    }
}

pub struct Subscribe<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    pub sender: Sender<NetMsg<T>>,
}

impl<T> Message for Subscribe<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T> Handler<Subscribe<T>> for Router
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: Subscribe<T>) -> Result<()> {
        assert!(self
            .subscribes
            .insert(T::ID, Box::new(msg.sender))
            .is_none());
        Ok(())
    }
}

pub struct SubscribeRpc<T>
where
    T: Message + HoneybeeMessage,
{
    pub caller: RpcCaller<T>,
}

impl<T> Message for SubscribeRpc<T>
where
    T: Message + HoneybeeMessage,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T> Handler<SubscribeRpc<T>> for Router
where
    T: Message + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SubscribeRpc<T>) -> Result<()> {
        assert!(self
            .subscribes
            .insert(T::ID, Box::new(msg.caller))
            .is_none());
        Ok(())
    }
}

pub struct SubscribeNats<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    pub sender: Sender<NatsMsg<T>>,
}

impl<T> Message for SubscribeNats<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T> Handler<SubscribeNats<T>> for Router
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: SubscribeNats<T>) -> Result<()> {
        assert!(self
            .subscribes
            .insert(T::ID, Box::new(msg.sender))
            .is_none());
        Ok(())
    }
}

pub struct NetMsg<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    pub source: Addr<TcpStreamActor>,
    pub msg: T,
}

impl<T> Message for NetMsg<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T> Handler<NetMsg<T>> for Router
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: NetMsg<T>) -> Result<()> {
        let sender = self
            .subscribes
            .get(&T::ID)
            .ok_or_else(|| anyhow!("route msg:{:?} not found handle actor", msg.msg))?;

        let sender = sender
            .downcast_ref::<Sender<NetMsg<T>>>()
            .ok_or_else(|| anyhow!("route msg:{:?} downcast_ref failed", msg.msg))?;

        sender.send(msg)
    }
}

pub struct NetRpcRouterMsg<T, R>
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    pub source: Addr<TcpStreamActor>,
    pub session: u32,
    pub msg: T,
}

impl<T, R> Message for NetRpcRouterMsg<T, R>
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T, R> Handler<NetRpcRouterMsg<T, R>> for Router
where
    T: Message<Result = R> + HoneybeeMessage,
    R: RpcResult,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: NetRpcRouterMsg<T, R>) -> Result<()> {
        if let Some(caller) = self.subscribes.get(&T::ID) {
            if let Some(caller) = caller.downcast_ref::<RpcCaller<T>>() {
                caller.rpc_call(msg.source, msg.session, msg.msg);
            } else {
                let _ = msg.source.send(SendMsg {
                    session: msg.session,
                    msg: RpcResponse::<R>(Err(RpcResponseError::RpcError(format!(
                        "downcast_ref failed, there is no remoter register this call process:{:?}",
                        msg.msg,
                    )))),
                });
            }
        } else {
            let _ = msg.source.send(SendMsg {
                session: msg.session,
                msg: RpcResponse::<T::Result>(Err(RpcResponseError::RpcError(format!(
                    "there is no remoter register this call process:{:?}",
                    msg.msg,
                )))),
            });
        }

        Ok(())
    }
}

pub struct NatsMsg<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    pub reply: String,
    pub msg: T,
}

impl<T> Message for NatsMsg<T>
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    type Result = ();
}

#[async_trait::async_trait]
impl<T> Handler<NatsMsg<T>> for Router
where
    T: Message<Result = ()> + HoneybeeMessage,
{
    async fn handle(&mut self, _ctx: &mut Context<Self>, msg: NatsMsg<T>) -> Result<()> {
        let sender = self
            .subscribes
            .get(&T::ID)
            .ok_or_else(|| anyhow!("route nats msg:{:?} not found handle actor", msg.msg))?;

        let sender = sender
            .downcast_ref::<Sender<NatsMsg<T>>>()
            .ok_or_else(|| anyhow!("route nats msg:{:?} downcast_ref failed", msg.msg))?;

        sender.send(msg)
    }
}

pub struct NetRpcMsg<T>
where
    T: Message + HoneybeeMessage,
{
    pub session: u32,
    pub source: Addr<TcpStreamActor>,
    pub msg: T,
}

impl<T> Message for NetRpcMsg<T>
where
    T: Message + HoneybeeMessage,
{
    type Result = <T as Message>::Result;
}

pub struct RouterContext<A> {
    pub router_addr: Addr<Router>,
    pub routee_addr: Addr<A>,
}

impl<A> RouterContext<A> {
    pub fn register<T>(&self)
    where
        T: Message<Result = ()> + HoneybeeMessage,
        A: Handler<NetMsg<T>>,
    {
        self.router_addr
            .send(Subscribe {
                sender: self.routee_addr.sender::<NetMsg<T>>(),
            })
            .ok();
    }

    pub fn register_rpc<T>(&self)
    where
        T: Message + HoneybeeMessage,
        A: Handler<NetRpcMsg<T>>,
        <T as Message>::Result: RpcResult,
    {
        self.router_addr
            .send(SubscribeRpc {
                caller: self.routee_addr.rpc_caller::<T>(),
            })
            .ok();
    }

    pub fn register_nats<T>(&self)
    where
        T: Message<Result = ()> + HoneybeeMessage,
        A: Handler<NatsMsg<T>>,
    {
        self.router_addr
            .send(SubscribeNats {
                sender: self.routee_addr.sender::<NatsMsg<T>>(),
            })
            .ok();
    }
}
